
#ifndef _TTG_PIPELINE_PackerNode_
#define _TTG_PIPELINE_PackerNode_
#include "Common/Definitions.h"

#include "Runtime.Low/Threads.h"
#include "Runtime.Low/Pipeline/Bindings.h"

#include "ProcessorNode.h"
#include "Port.h"
#include "TaskKernels.h"

namespace ttg_internal
{
	namespace pipeline
	{
		//Set, that contains serial tasks.
		class SerialTaskSet
		{
			private:
				std::vector<ttg::services::WiseTask *> tasks;	//single threaded tasks.
				bool is_performing;								//is single threaded task performing?
				ttg::threads::FastCriticalSection locker;		//locker for collection.
			public:
				//Creates empty set for serial tasks.
				inline SerialTaskSet()
				{ is_performing = false; }
				//Adds task for signle threading execution.
				//Returns task for execution (or NULL).
				inline ttg::services::WiseTask *add(ttg::services::WiseTask *task)
				{
					bool should_perform = false;
					locker.enter();
					if (!is_performing)
					{
						should_perform = true;
						is_performing = true;
					}
					else
						tasks.push_back(task);
					locker.leave();
					//Returning.
					return should_perform ? task : NULL;
				}
				//Sends another serial task to execution (if there is such task).
				//Returns task for execution (or NULL).
				inline ttg::services::WiseTask *launchTask()
				{
					//Getting another task.
					ttg::services::WiseTask *cur = NULL;
					locker.enter();
					if (tasks.size() != 0)
					{
						cur = tasks[tasks.size() - 1];
						tasks.pop_back();
					}
					else
						is_performing = false;
					locker.leave();
					//Returning.
					return cur;
				}
		};
		//Responds for data packing into tasks.
		//Also manages bindings.
		template <class InType, class OutType,
				  class ProcessInType, class ProcessOutType>
		class PackerNode :public ProcessorNode<InType, OutType, ProcessInType, ProcessOutType>
		{
			private:
				ttg::pipeline::ContextMap<SerialTaskSet> serial_tasks;
				std::vector<ttg_internal::pipeline::Binding<InType,
															OutType,
															ProcessInType,
															ProcessOutType> *> bindings;
			protected:
				//Adds binding to special device.
				//Note: it will be deleted by destructor.
				inline void addBinding(ttg_internal::pipeline::Binding<InType,
																	   OutType,
																	   ProcessInType,
																	   ProcessOutType> *binding)
				{
					bindings.push_back(binding);
					binding->setOwner(this);
				}
				//Returns all bindings.
				inline std::vector<ttg_internal::pipeline::Binding<InType,
																   OutType,
																   ProcessInType,
																   ProcessOutType> *> &getBindings()
				{ return bindings; }
				//ProcessorNode member.
				virtual void registerTask(ttg::services::WiseTask *task,
										  ttg::pipeline::ContextID ID,
										  double abstract_difficulty)
				{
					//Setting handler.
					task->onFinish() += make_delegate(this, taskHasBeenPerformed);
					//Updating state.
					getState(ID).getLocker().enter();
					getState(ID).getActiveTaskCount() += 1;
					getState(ID).getLocker().leave();
					//Updating informations.
#ifdef TTG_PL_VISUALIZER
						getInfo().getLocker().enter();
						getInfo().getTaskCount() += 1;
						getInfo().getActiveTaskCount() += 1;
						getInfo().getAbstractDifficulty() += abstract_difficulty;
						getInfo().getLocker().leave();
#endif
				}
				//Generates task for data processing.
				inline ttg::services::WiseTask *createAndRegisterTask(ProcessInType data,
																	  ttg::pipeline::ContextID ID)
				{
					//Creating task.
					double dif = getDifficulty(data);
					ttg::services::WiseTask *res
						= new ttg::services::WiseTask(getInfo().getCoefficient() * dif);
					res->addKernel(ttg::devices::CPU::x86::getType(),
								   new ProcessKernel<ProcessInType>(this, data, ID));
					//Adding bindings.
					for (size_t i = 0; i < bindings.size(); i++)
					{
						if (bindings[i]->getDifficulty(data) >= 0.0)
							res->addKernel(bindings[i]->getTargetDevice(),
										   new ProcessKernel<ProcessInType>(bindings[i], data, ID));
					}
					//Registering and returning.
					registerTask(res, ID, dif);
					return res;
				}
				//Creates tasks for data processing [thread-safe].
				//If there aren't ready data, returns NULL.
				std::vector<ttg::services::WiseTask *> createAndRegisterTasks(std::vector<ProcessInType> &data,
																			  ttg::pipeline::ContextID ID)
				{
					std::vector<ttg::services::WiseTask *> res;
					for  (size_t i = 0; i < data.size(); i++)
					{
						//Creating task.
						ttg::services::WiseTask *cur = createAndRegisterTask(data[i], ID);
						//Sending to serial tasks or to result vector.
						if (getInfo().isSingleThreaded())
						{
							ttg::services::WiseTask *t = serial_tasks[ID].add(cur);
							if (t != NULL)
								getIContext().getTaskGatherer(ID).addTask(t);
						}
						else
							res.push_back(cur);
					}
					return res;
				}
				//Packs all buffer of current thread into tasks.
				inline std::vector<ttg::services::WiseTask *> packBuffers(ttg::pipeline::ContextID ID)
				{
					std::vector<ttg::services::WiseTask *> res;
					//Packing "sendThis()" buffer.
					std::vector<ttg::services::WiseTask *> &this_tasks =
							packBuffer(getIContext().getDataBuffer().getThisBuffer(), ID);
					getIContext().getDataBuffer().getThisBuffer().clear();
					res.insert(res.end(), this_tasks.begin(), this_tasks.end());
					//Packing "sendNext()" buffer.
					for (size_t i = 0; i < getParametrizedOutputs().size(); i++)
					{
						ParametrizedOutput<OutType> *output = getParametrizedOutputs()[i];
						ttg::structures::UniversalStack stack
							= getIContext().getDataBuffer().getNextBuffer();
						std::vector<ttg::services::WiseTask *> &next_tasks = output->packBuffer(stack, ID);
						res.insert(res.end(), next_tasks.begin(), next_tasks.end());
					}
					getIContext().getDataBuffer().getNextBuffer().clear();
					//Returning result.
					return res;
				}
				//IDataProcessor<> member.
				virtual void startProcessing(ttg::pipeline::ContextID ID)
				{
					getIContext().getDataBuffer().getContextID() = ID;
				}
				//IDataProcessor<> member.
				virtual void finishProcessing(ttg::pipeline::ContextID ID)
				{
					getIContext().getTaskGatherer(ID).addTasks(packBuffers(ID));
				}
				//Sends another serial task to execution (if there is such task).
				inline void taskHasBeenPerformed_LaunchSerialTask(ttg::pipeline::ContextID ID)
				{
					//Getting task.
					ttg::services::WiseTask *task = serial_tasks[ID].launchTask();
					//Executing it.
					//We should not update active_tasks field
					//due to it was updated by task creator.
					if (task != NULL)
						getIContext().getTaskGatherer(ID).addTask(task);
				}
				//Updates informations. Should be called after each performed task.
				inline void taskHasBeenPerformed_UpdateInfo(double time,
															ttg::pipeline::ContextID ID)
				{
					//Updating info.
					getInfo().getLocker().enter();
					getInfo().getTimeDifficulty() += time;
#ifdef TTG_PL_VISUALIZER
					//Resizing.
					size_t thread_id = getIContext().getPerformers().getIndex();
					if (thread_id >= getInfo().getThreadUsage().size())
					{
						size_t old_size = getInfo().getThreadUsage().size();
						size_t new_size = thread_id + 1;
						std::vector<double> &en = getInfo().getThreadUsage();
						en.resize(new_size);
						for (size_t i = old_size; i < new_size; i++)
							en[i] = 0.0;
					}
					getInfo().getThreadUsage()[thread_id] += time;
					getInfo().getActiveTaskCount() -= 1;
#endif
					getInfo().getLocker().leave();
					//Updating state.
					getState(ID).getLocker().enter();
					getState(ID).getActiveTaskCount() -= 1;
					getState(ID).getLocker().leave();
				}
				//Checks completeness and changes node state.
				virtual void taskHasBeenPerformed_CheckCompleteness(ttg::pipeline::ContextID ID) = 0;
				//Checks operability of this node (e.g. is there specific output).
				//Will be called before any launching. In case of error should throw Error().
				virtual void checkOperability()
						{ /*nothing*/ }
			public:
				//Creates node with given name.
				inline PackerNode(const char *name, bool is_serial)
					:ProcessorNode(name, is_serial)
				{ /*nothing*/ }
				virtual ~PackerNode()
				{ SAFE_DELETE_CONTENT(&bindings); }
				//Handler for performed tasks.
				static void taskHasBeenPerformed(PackerNode<InType, OutType,
														ProcessInType, ProcessOutType> *object,
												 ttg::services::WiseTask *message)
				{
					//Getting ID of context.
					ttg::pipeline::ContextID ID
						= object->getIContext().getDataBuffer().getContextID();
					//Notifying.
					if (object->getInfo().isSingleThreaded())
						object->taskHasBeenPerformed_LaunchSerialTask(ID);
					object->taskHasBeenPerformed_UpdateInfo(message->getElapsedTime(), ID);
					object->taskHasBeenPerformed_CheckCompleteness(ID);
					//Updating info about performers.
#ifdef TTG_PL_VISUALIZER
					object->getIContext().getPerformers().getValue()
								.addTaskInfo(PipelineTaskInfo(message->getStartTime(),
															  message->getElapsedTime(),
															  message->getExpectedTime()));
#endif
				}
		};
	}
}

#endif
